package com.xnyzc.lhy.nio.netty;

import com.alibaba.fastjson.JSON;
import com.xnyzc.lhy.common.cache.NettyChannelMap;
import com.xnyzc.lhy.common.constant.MethodConstant;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

/**
 * 初始化Netty服务
 *
 * @author liuju
 * @create 2019-07-17 13:23
 */
@Slf4j
@Component
@ChannelHandler.Sharable
public class ServerHandler extends ChannelInboundHandlerAdapter {

    private AttributeKey<ChannelDelayed> KEY_UNCHANNEL = AttributeKey
            .valueOf("ucid");

    private DelayQueue delayQueue = new DelayQueue();

    // FIXME 添加TCP连接缓存

    @Autowired
    private NettyServerService serverService;

    public ServerHandler() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    ChannelDelayed element = null;
                    try {
                        element = (ChannelDelayed) delayQueue.poll(1, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    if (element != null) {
                        element.run();
                    }
                }
            }
        }).start();
    }

    /**
     * 当前channel从远端读取到数据
     *
     * @param ctx
     * @param msg
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        Channel channel = ctx.channel();
        InetSocketAddress inSocket = (InetSocketAddress) channel.remoteAddress();
        // log.info(inSocket.getAddress() + ": 收到客户端数据.......");
        ChannelDelayed delayed = channel.attr(KEY_UNCHANNEL).get();
        if (delayed == null) {
            try {
                // 调用service
                serverService.judgeOperation(ctx, msg);
            } catch (Exception e) {
                log.error("当前channel从远端读取到数据: {}", e);
            } finally {
                ReferenceCountUtil.release(msg);
            }
        } else {
            //认证注册信息
            if (serverService.regist(ctx, msg)) {
                channel.attr(KEY_UNCHANNEL).set(null);
            } else {
                ctx.close();
            }
        }
    }


    /**
     * 在连接被建立并且准备进行通信时被调用
     */
    @Override
    public void channelActive(final ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        InetSocketAddress inSocket = (InetSocketAddress) channel.remoteAddress();
        delayQueue.add(new ChannelDelayed(channel));
        log.info(inSocket.getAddress() + ": 连接成功......");
    }

    /**
     * 心跳，用户事件触发的时候
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        log.info("发送心跳...");
        String msg = NettyMsg.setJsonMsg(MethodConstant.PING, JSON.toJSONString(new HashMap<String, Object>(16)));
        ctx.writeAndFlush(Unpooled.copiedBuffer((msg).getBytes("GBK")));
        super.userEventTriggered(ctx, evt);
    }

    /**
     * 连接处于不活跃时调用（连接关闭），当前channel不活跃的时候，也就是当前channel到了它生命周期末
     **/
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        InetSocketAddress inSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        log.info(inSocket.getAddress() + ": 断开连接......");
        super.channelInactive(ctx);
        NettyChannelMap.remove(ctx);
    }

    /**
     * channel read消费完读取的数据的时候被触发
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    /**
     * 处理方法是当出现Throwable对象才会被调用
     **/
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }

    public class ChannelDelayed implements Delayed {
        /**
         * 默认延时3秒
         */
        public static final long DEFAULT_DELAY = 3 * 1000;

        private Channel channel;
        private long startTime;

        public ChannelDelayed(Channel channel) {
            this(channel, DEFAULT_DELAY);

            channel.attr(KEY_UNCHANNEL).set(this);
        }

        public ChannelDelayed(Channel channel, long delayTime) {
            this.channel = channel;
            this.startTime = System.currentTimeMillis() + delayTime;
        }

        public void run() {
            ChannelDelayed mchannel = channel.attr(KEY_UNCHANNEL).get();
            if (mchannel != null && mchannel == this && channel.isActive()) {
                log.info("断开连接");
                channel.close();
            }
        }

        public long getDelay(TimeUnit unit) {
            return unit.convert(this.startTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed delayed) {
            ChannelDelayed channelDelayed = (ChannelDelayed) delayed;
            if (startTime > channelDelayed.startTime) {
                return 1;
            } else {
                return -1;
            }
        }
    }
}
